package com.lunaler.stream

import java.util.concurrent.atomic.AtomicInteger

import com.alibaba.fastjson.JSON
import com.lunaler.stream.{AdsUvLog, JdbcUtils}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

object LogMain {
  def main(args: Array[String]): Unit = {
    var conf = new SparkConf().setMaster("local[5]").setAppName("kafka")

    var sc = new SparkContext(conf)

    var ssc = new StreamingContext(sc,Seconds(2))


    var zkHost = "172.31.20.59:2181"

    var groupId = "gp01"

    //var topic = Map("topic_event"->2)
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "172.31.20.59:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "gp01",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("topic_event")
    val kafkaSource = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    val lines = kafkaSource.map(_.value)
    val index = new AtomicInteger();
    lines.foreachRDD(rdd=>{
            val lines=rdd.toLocalIterator

            //遍历迭代器,指定对每条数据的处理逻辑
            while(lines.hasNext) {
              try {
                val line = lines.next()
                val arr = line.split("\\|")
                val time = arr(0)
                val app = JSON.parseObject(arr(1)).get("ap").toString
                JdbcUtils.saveOrUpdate(time.toLong, app)
//                if (index.incrementAndGet() > 14000)
//                println(index.incrementAndGet());
              } catch {
                case ex: Exception =>{
                  //donothing
                }
              }
            }
    })

    ssc.start()
    ssc.awaitTermination()

  }
}
